package mq.rabbit.demo;

import com.google.common.collect.Lists;
import com.rabbitmq.client.*;

import mq.rabbit.common.util.RabbitMQUtils;
import org.junit.Test;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Routing之订阅模型-Direct（直连）:
 * <p>
 * <br />
 * <p>
 * 在Fanout（广播）模式中，一条消息，会被所有订阅的队列都消费。
 * 但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
 * <p>
 * <br />
 * <p>
 * 在Direct模型下：
 * <br />
 * <li>队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）</li>
 * <li>消息的发送方在向Exchange发送消息时，也必须指定消息的RoutingKey</li>
 * <li>Exchange不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey进行判断，
 * 只有队列的RoutingKey与消息的RoutingKey完全一致，才会接收到消息</li>
 * <p>
 * <br />
 * <li>P：生产者：向Exchange发送消息，发送消息时，会指定一个RoutingKey</li>
 * <li>X：Exchange（交换机）：接收生产者的消息，然后把消息递交与RoutingKey完全匹配的队列</li>
 * <li>C1：消费者1：其所在队列指定了需要RoutingKey为error的消息</li>
 * <li>C2：消费者2：其所在队列指定了需要RoutingKey为info、error、waring的消息</li>
 */
public class Routing_Direct {

    @Test
    public void testProvider() throws IOException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConnection();

        // 获取通道
        Channel channel = connection.createChannel();

        String exchangeName = "logs_direct";

        // 通过通道声明交换机
        // 参数1：交换机名称
        // 参数2：交换机类型  direct表示路由模式
        channel.exchangeDeclare(exchangeName, "direct");

        // 发送消息
        String routingKey = "error";
//        String routingKey = "info";
        String message = "这是direct模型发布的基于routerKey: [" + routingKey + "] 发送的消息";
        channel.basicPublish(exchangeName, routingKey, null, message.getBytes());

        // 关闭
        RabbitMQUtils.closeChanelAndConnection(channel, connection);
    }

    /**
     * 基于routerKey绑定队列和交换机，只接收对应的 routingKey 消息。
     *
     * @throws IOException
     * @throws InterruptedException
     */
    @Test
    public void testConsumer() throws IOException, InterruptedException {

        List<List<String>> routingKeys = Lists.newArrayList(Lists.newArrayList("error"), Lists.newArrayList("info", "error", "waring"));

        for (List<String> routingKeyList : routingKeys) {

            Connection connection = RabbitMQUtils.getConnection();
            Channel channel = connection.createChannel();

            String exchangeName = "logs_direct";

            // 通道声明交换机以及交换机的类型
            channel.exchangeDeclare(exchangeName, "direct");

            // 创建一个临时队列
            String queueName = channel.queueDeclare().getQueue();

            // 基于routerKey绑定队列和交换机
            for (String routingKey : routingKeyList) {
                channel.queueBind(queueName, exchangeName, routingKey);
            }

            // 获取消息
            channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者 " + queueName + " >>>>> " + new String(body));
                }
            });

        }
        TimeUnit.MINUTES.sleep(2);
    }
}
